use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;
use chrono::Local;
use fudata::db;
use fudata::db::todo;
use fudata::model;
use log::warn;
use tokio::sync::Mutex;
use tokio::sync::Semaphore;
use crate::todo::worker_sse::sse_worker;
use crate::todo::MAX_KZZ_SECURITY;
use fudata::model::todo::Todo;
use fudata::model::todo::TodoStatus;
use fudata::model::todo::TodoType;
use log::error;
use log::info;
use crate::helper;

use super::evenly_hosts::M001KLINES_HOST_MANAGER;
use super::worker_sse::EastmoneySseType;
use super::worker_sse::SseClientWrapper;
use super::worker_sse::SseEventInjector;

///
/// every day we update the kzz list after 15:30
///
pub async fn generate_todo_sse_kzz_minute1klines() -> Result<()> {
    let start = Local::now();

    info!("generate_todo_sse_kzz_minute1klines start...");

    let rs = db::security::select(
        &None,
        Some(model::security::SecurityType::KZZ),
        None,
        None,
        0,
        MAX_KZZ_SECURITY
    ).await;
    match rs {
        Ok(securities) => {
            let mut _i = 0;
            for security in securities {
                let security_id = &security.id_raw();
                let todo_id = Todo::id_from(
                    &[security_id, TodoType::SseM001Klines.to_string().as_str()]
                );

                let todo_rs = todo::find(todo_id.as_str()).await;
                match todo_rs {
                    Ok(v) =>
                        match v {
                            Some(todo) => {
                                // if there is pending todo, then let it be or else update the plan_at to tomorrow
                                if todo.is_done() || todo.is_failed() {
                                    let todo_update = Todo {
                                        // updatable fields
                                        plan_at: Some(helper::now_plus_date_time(0, 0, 0, 0)),
                                        status: Some(TodoStatus::Pending),
                                        ..Default::default()
                                    };
                                    let _ = todo::upsert(todo_id.as_str(), &todo_update).await?;
                                    print!("U");
                                }
                            }
                            None => {
                                let todo_new = Todo {
                                    name: Some(
                                        format!("订阅可转债证券{}的1分钟K线数据", security_id)
                                    ),
                                    type_: Some(TodoType::SseM001Klines),
                                    // updatable fields
                                    plan_at: Some(helper::now_plus_date_time(0, 0, 0, 0)),
                                    status: Some(TodoStatus::Pending),
                                    created_at: Some(Local::now()),
                                    ..Default::default()
                                };
                                let _ = todo::upsert(todo_id.as_str(), &todo_new).await?;
                                print!("N");
                            }
                        }
                    Err(e) => error!("{}", e),
                }
            }
        }
        Err(e) => error!("{}", e),
    }
    info!(
        "generate_todo_sse_kzz_minute1klines end/ spent seconds={} ",
        Local::now().signed_duration_since(start).num_seconds()
    );
    Ok(())
}

pub async fn sse_minute1klines(
    worker_no: u8,
    todo: &Todo,
    sse_event_injector: Arc<SseEventInjector>,
    sse_workers: Arc<Mutex<HashMap<String, tokio::task::JoinHandle<()>>>>,
    semaphore: Arc<Semaphore>,
    connection_pool: Arc<Mutex<HashMap<String, SseClientWrapper>>>
) {
    let security_id_rs = db::todo::get_todo_string_parameter_by(todo, "security_id");
    let eastmoney_security_id_rs = db::todo::get_todo_string_parameter_by(
        todo,
        "eastmoney_security_id"
    );

    if
        let (Ok(security_id), Ok(eastmoney_security_id)) = (
            &security_id_rs,
            &eastmoney_security_id_rs,
        )
    {
        let mut sse_workers_lock = sse_workers.lock().await;
        if let None = sse_workers_lock.get(eastmoney_security_id) {
            let host = M001KLINES_HOST_MANAGER.add_to_hosts_evenly(eastmoney_security_id);

            let rs = get_or_create_minute1klines_connection(
                security_id,
                eastmoney_security_id,
                &host,
                connection_pool
            ).await;
            match rs {
                Ok(client_wrapper) => {
                    let task: tokio::task::JoinHandle<()> = tokio::spawn(async move {
                        let rs = sse_worker(
                            worker_no as u16,
                            &client_wrapper,
                            sse_event_injector,
                            semaphore.clone()
                        ).await;

                        if let Err(e) = rs {
                            log::error!("(sse_minute1klines) Error in sse_worker: {:?}", e);
                        }
                    });

                    info!("(sse_minute1klines) started task {:?}", task);
                    sse_workers_lock.insert(eastmoney_security_id.to_string(), task);
                }
                Err(e) => error!("(sse::init) failed to create connection{:?}", e),
            }
        };
    }
}

///
/// @security_id, e.g. 1.600519
///
/// e.g. "https://17.push2.eastmoney.com/api/qt/stock/trends2/sse?fields1=f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13,f17&fields2=f51,f52,f53,f54,f55,f56,f57,f58&mpi=1000&ut=fa5fd1943c7b386f172d6893dbfba10b&secid=1.600519&ndays=1&iscr=0&iscca=0&wbp2u=|0|0|0|web"
///
/// @host, e.g. 24.push2.eastmoney.com, 25.push2.eastmoney.com
///
pub async fn get_or_create_minute1klines_connection(
    security_id: &str,
    eastmoney_security_id: &str,
    host: &str,
    connection_pool: Arc<Mutex<HashMap<String, SseClientWrapper>>>
) -> Result<SseClientWrapper> {
    let mut pool = connection_pool.lock().await;
    let exist_client_wrapper = pool.get(&eastmoney_security_id.to_string());
    match exist_client_wrapper {
        Some(v) => {
            warn!("(get_or_create_tick3_connection) found existing connection for {}", eastmoney_security_id);
            Ok(v.clone())
        }
        None => {
            // "https://17.push2.eastmoney.com/api/qt/stock/trends2/sse?fields1=f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13,f17&fields2=f51,f52,f53,f54,f55,f56,f57,f58&mpi=1000&ut=fa5fd1943c7b386f172d6893dbfba10b&secid=1.600519&ndays=1&iscr=0&iscca=0&wbp2u=|0|0|0|web"
            // Payload: Query String Parameters
            // fields1: f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13,f17
            // fields2: f51,f52,f53,f54,f55,f56,f57,f58
            // mpi: 1000
            // ut: fa5fd1943c7b386f172d6893dbfba10b
            // secid: 1.600519
            // ndays: 1
            // iscr: 0
            // iscca: 0
            // wbp2u: |0|0|0|web

            let ut = utils::generate_random_hex_string(32);
            let url = format!(
                "https://{host}/api/qt/stock/trends2/sse?fields1=f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13,f17&fields2=f51,f52,f53,f54,f55,f56,f57,f58&mpi=1000&ut={ut}&secid={eastmoney_security_id}&ndays=1&iscr=0&iscca=0&wbp2u=|0|0|0|web"
            );

            let es_builder = eventsource_client::ClientBuilder
                ::for_url(&url)?
                // Set custom headers if needed
                .header("Accept", "text/event-stream")?
                .header("Accept-Encoding", "gzip, deflate, br, zstd")?
                .header("Accept-Language", "zh-CN,zh;q=0.9,en;q=0.8")?
                .header("Cache-Control", "no-cache")?
                .header("Connection", "keep-alive")?
                .header("Host", &format!("{host}"))?
                .header("Origin", "https://quote.eastmoney.com")?
                .header("Pragma", "no-cache")?
                .header("Referer", "https://quote.eastmoney.com/sh513130.html")?
                .header("Sec-Fetch-Dest", "empty")?
                .header("Sec-Fetch-Mode", "cors")?
                .header("Sec-Fetch-Site", "same-site")?
                .header(
                    "User-Agent",
                    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
                )?
                .header(
                    "sec-ch-ua",
                    "\"Not/A)Brand\";v=\"8\", \"Chromium\";v=\"126\", \"Google Chrome\";v=\"126\""
                )?
                .header("sec-ch-ua-mobile", "?0")?
                .header("sec-ch-ua-platform", "\"macOS\"")?;

            let client = es_builder
                .reconnect(
                    eventsource_client::ReconnectOptions
                        ::reconnect(true)
                        .retry_initial(false)
                        .delay(Duration::from_secs(1))
                        .backoff_factor(2)
                        .delay_max(Duration::from_secs(60))
                        .build()
                )
                .build();

            let client_wrapper = pool
                .entry(eastmoney_security_id.to_string())
                .or_insert_with(|| {
                    SseClientWrapper::new(
                        security_id,
                        eastmoney_security_id,
                        host,
                        EastmoneySseType::MINUTE1,
                        Arc::new(client)
                    )
                });

            // Clone the Arc inside the ClientWrapper to create a new ClientWrapper instance.
            Ok(client_wrapper.clone())
        }
    }
}

#[cfg(test)]
mod test {
    use fudata::db;
    use fudata::model::todo::TodoType;
    use fudata::model::todo::Todo;

    use crate::todo::sse_minute1klines_todo;

    #[tokio::test]
    async fn test_generate_todo_crawl_kzz_list() {
        pretty_env_logger::init_timed();
        let _ = fudata::db::init().await.unwrap();

        let _ = sse_minute1klines_todo::generate_todo_sse_kzz_minute1klines().await;

        let todo_id = Todo::id_from(
            &["SH.KZZ.110052", TodoType::SseM001Klines.to_string().as_str()]
        );

        let rs = db::todo::find(todo_id.as_str()).await;
        println!("rs: {:?}", rs);
    }
}
